package com.kingjw.windowsapi

import com.kingjw.sourceTest.SensorReading
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable

/**
 *
 * @Package: com.kingjw.windowsapi
 * @ClassName: TumblingEventTimeWindows
 * @Author: 王坚伟
 * @CreateTime: 2022/1/19 15:52
 * @Description:
 */
object TumblingEventTimeWindowsTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val dstream: DataStream[String] = env.socketTextStream("10.0.3.207", 7777)
    val textWithTsDstream: DataStream[(String, Long, Int)] = dstream.map { text =>
      val arr: Array[String] = text.split(",")
      (arr(0).trim, arr(1).trim.toLong, 1)
    }
    val textWithEventTimeDstream: DataStream[(String, Long, Int)] =
      textWithTsDstream.assignTimestampsAndWatermarks(new
          BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.milliseconds(1000)) {
        override def extractTimestamp(element: (String, Long, Int)): Long = {
          return element._2
        }
      })

    val textKeyStream = textWithEventTimeDstream.keyBy(0)
    textKeyStream.print("textkey:")

    val windowStream = textKeyStream.window(TumblingEventTimeWindows.of(Time.seconds(2)))
    val groupDstream: DataStream[mutable.HashSet[Long]] =
      windowStream.fold(new mutable.HashSet[Long]()) { case (set, (key, ts, count)) =>
        set += ts
      }
    groupDstream.print("window::::").setParallelism(1)
    env.execute()
  }

}

class TemperatureAlertFunction(val threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {
  private var lastTempState: ValueState[Double] = _

  override def open(parameters: Configuration): Unit = {
    val lastTempDescriptor = new ValueStateDescriptor[Double]("lastTemp", classOf[Double])
    lastTempState = getRuntimeContext.getState[Double](lastTempDescriptor)
  }

  override def flatMap(reading: SensorReading, out: Collector[(String, Double, Double)]): Unit = {
    val lastTemp = lastTempState.value()
    val tempDiff = (reading.temperature - lastTemp).abs
    if (tempDiff > threshold) {
      out.collect((reading.id, reading.temperature, tempDiff))
    }
    this.lastTempState.update(reading.temperature)
  }
}
